package com.laher.demo.dao;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.joda.time.DateTime;

import com.laher.demo.mapreduce.TwitsTableMap;
import com.laher.demo.model.Twit;
import com.laher.demo.model.User;

public class TwitsDao {
	public static final byte[] TABLE_NAME=Bytes.toBytes("twits");
	public static final byte[] FAM_INFO=Bytes.toBytes("twits");
	
	public static final byte[] COL_USER=Bytes.toBytes("user");
	public static final byte[] COL_TWIT=Bytes.toBytes("twit");

	private Table table;
	
	public TwitsDao(Connection connection) throws IOException{
		this.table = connection.getTable(TableName.valueOf(TABLE_NAME));
	}
	
	private static Get mkGet(String user){
		Get get = new Get(Bytes.toBytes(user));
		get.addFamily(FAM_INFO);
		return get;
	}
	
	private static Put mkPut(TwitMap t){
		Put p = new Put(Bytes.toBytes(t.user));
		p.addColumn(FAM_INFO, COL_USER, Bytes.toBytes(t.user));
		p.addColumn(FAM_INFO, COL_TWIT, Bytes.toBytes(t.text));
		return p;
	}
	
	private static Delete mkDel(String user){
		Delete del = new Delete(Bytes.toBytes(user));
		return del;
	}
	
	public void addTwit(String user, DateTime dt, String text) throws IOException{
		Put put = mkPut(new TwitMap(user, dt, text));
		table.put(put);
		table.close();
	}
	
	public void deleteUser(String user) throws IOException {
		table.delete(mkDel(user));
		table.close();
	}
	
	public List<Twit> getTwits() throws IOException{
		List<Twit> ts = new ArrayList<Twit>();
		ResultScanner rs = table.getScanner(new Scan());
		for (Result r : rs) {
			ts.add(new TwitMap(r));
		}
		return ts;
	}
	
	public String getJob() throws Exception {
		Job job = Job.getInstance(new JobConf(table.getConfiguration()),"TimeSpent");
		TableMapReduceUtil.initTableMapperJob("twits", new Scan(), TwitsTableMap.class, ImmutableBytesWritable.class, Result.class, job);
		
//		job.waitForCompletion(true);
//		job.setOutputFormatClass(TextOutputFormat.class);
//		job.setInputFormatClass(TextInputFormat.class);
		
	    job.setOutputFormatClass(NullOutputFormat.class);
	    job.setNumReduceTasks(0);
	    System.out.println("========>"+(job.waitForCompletion(true) ? 0 : 1));
		return job.getJobFile();
	}

	public void postTwit(String user, DateTime dt, String text) throws IOException{
		Put p = mkPut(new TwitMap(user, dt, text));
		table.put(p);
		table.close();
	}
	
	private static class TwitMap extends Twit{
		private TwitMap(Result result) {
			this(result.getValue(FAM_INFO, COL_USER),
				null,
				result.getValue(FAM_INFO, COL_TWIT));
		}
		private TwitMap(byte[] user, byte[] dt, byte[] text) {
			this(
	           Bytes.toString(user),
	           dt==null?null:new DateTime(-1 * Bytes.toLong(dt)),
	           Bytes.toString(text));
		}
		private TwitMap(String user,DateTime dt,String text){
			this.user = user;
			this.dt = dt;
			this.text = text;
		}
	}
}
